package yahoodata;

import java.io.*;
import java.lang.*;
import java.net.*;
import java.text.*;
import java.util.*;

/** There is nothing OOP about this worker. He just takes care of business in the most efficient way possible:
 *  (1) wait of a symbol to be inserted into a synchronized Map.
 *  (2) read, and then delete, a symbol from the synchronized Map.
 *  (3) connect to Yahoo Finance and download/store the trade event data into a List.
 *  (4) write the trade event to a file using a synchronized FileWriter.
 *  (5) create a summary view of the trade event data.
 *  (6) write the summary view to a file using a different synchronized FileWriter.
 *  (7) go back to step #1 and wait for another symbol to be inserted into the Map.
 *  (8) [exit condition]: exit when the worker is notified that there are no more symbols to be
 *                        inserted into the symbol map, and also the symbol Map is empty.
 *  
 *  ___format of the trade event file___
 *  column #01 = symbol
 *  column #02 = trade date

 *
 *  ___format of the summary event view file___
 *  column #01 = symbol
 *  column #03 = start date
 *  column #04 = end   date
 *
 */
public class Worker implements Runnable {
    public        Thread.State     state = Thread.State.NEW;
    private       int              initialAmountOfWork = -1;
    private       int              workCompleted = 0;
    private final String           id;
    private final String           m1, d1, y1;
    private final String           m2, d2, y2;    
    private final Set<String>      workQueue;
    private final String           eventStartDateString,   eventEndDateString;
    private final String           profileStartDateString, profileEndDateString;
    private final Main.ProductType productType;
    private final BufferedWriter   eventWriter, profileWriter, errorWriter;
    
    public Worker(String         eventStartDateString,  String         eventEndDateString,  String           profileStartDateString,
                  String         profileEndDateString,  BufferedWriter profileWriter,       BufferedWriter   eventWriter,
                  BufferedWriter errorWriter,           Set<String>    workQ,               Main.ProductType productType, String id) {
        this.profileWriter = profileWriter;
        this.eventWriter   = eventWriter;
        this.errorWriter   = errorWriter;
        
        this.workQueue   = workQ;
        this.productType = productType;
        this.id          = id;
        
        this.eventStartDateString   = eventStartDateString;
        this.eventEndDateString     = eventEndDateString;
        this.profileStartDateString = profileStartDateString;
        this.profileEndDateString   = profileEndDateString;   
        
        StringTokenizer split = new StringTokenizer(eventStartDateString, "-");
        m1 = (String) Integer.toString(Integer.valueOf(split.nextToken()) - 1);  // the month format range is (0 -> 11). (Jan. -> December)
        d1 = split.nextToken();
        y1 = split.nextToken();
        split = new StringTokenizer(eventEndDateString, "-");
        m2 = (String) Integer.toString(Integer.valueOf(split.nextToken()) - 1); // the month format range is (0 -> 11). (Jan. -> December)
        d2 = split.nextToken();
        y2 = split.nextToken();
    }
    
    
    public void run() {
        state = Thread.State.RUNNABLE;
        System.out.println("____worker id#" + id + "_::___starting....");
        
        NumberFormat priceFormat = NumberFormat.getInstance();
        priceFormat.setMaximumFractionDigits(2);
        priceFormat.setMinimumFractionDigits(2);
        
        NumberFormat volatilityFormat4 = NumberFormat.getInstance();
        volatilityFormat4.setMaximumFractionDigits(4);
        volatilityFormat4.setMinimumFractionDigits(4);
        
        NumberFormat volatilityFormat6 = NumberFormat.getInstance();
        volatilityFormat6.setMaximumFractionDigits(6);
        volatilityFormat6.setMinimumFractionDigits(6);
        
        NumberFormat volumeFormat = NumberFormat.getInstance();
        volumeFormat.setMaximumFractionDigits(0);
        volumeFormat.setMinimumFractionDigits(0);
       
                
        while(true) {
            try {
                List<TradeEvent> timeSeries = new ArrayList<TradeEvent>();
                String           symb       = null;
                
                synchronized(workQueue) {
                    while(true) {
                        if(workQueue.size() > 0) {
                            if(initialAmountOfWork < 0) { initialAmountOfWork = workQueue.size(); }
                            symb = (String) workQueue.toArray()[0];
                            workQueue.remove(symb);
                            workQueue.notify();
                            state = Thread.State.RUNNABLE;
                            break;
                        } else {
                            state = Thread.State.WAITING;
                            try { workQueue.wait(); } catch(InterruptedException e) { }
                            if(workQueue.size() == 0) {
                                state = Thread.State.TERMINATED;
                                return;
                            }
                        }
                    }
                }
    
                URL url = this.createYahooURL(symb);
                try {
                    BufferedReader urlReader         = new BufferedReader(new InputStreamReader(url.openStream()));
                    boolean        rowHeadersSkipped = false;
                    while(true) {
                        String inLine = (String) urlReader.readLine();
                        if(inLine == null) { break; }
                                                     // notice: the very first line downloaded is not numeric.
                                                     // rather it is the columns header: "Date,Open,High,Low,Close,Volume,AdjClose".
                        if(rowHeadersSkipped) {
                            timeSeries.add(new TradeEvent(symb, inLine, timeSeries));
                        } else { rowHeadersSkipped = true; }
                    }
                    urlReader.close();
                } catch(FileNotFoundException e) {   // input files often have symbols not valid for Yahoo Finance.
                        synchronized(errorWriter) {
                            errorWriter.newLine();
                            errorWriter.write(symb);
                        }
                        initialAmountOfWork--;
                        continue;
                    }

                synchronized(eventWriter) {
                    for(int i = 0; i < timeSeries.size(); i++) {
                        TradeEvent te = (TradeEvent) timeSeries.get(i);
                        StringBuilder outLine = new StringBuilder(te.symb + ";");
                        outLine.append(te.date                      + ";");
                        outLine.append(priceFormat.format(te.open)  + ";");
                        outLine.append(priceFormat.format(te.high)  + ";");
                        outLine.append(priceFormat.format(te.low)   + ";");
                        outLine.append(priceFormat.format(te.close) + ";");
                        outLine.append(volumeFormat.format(te.volume).replaceAll(",", "")    + ";");
                        outLine.append(priceFormat.format(te.adjClose)                       + ";");
                        outLine.append(volumeFormat.format(te.dolVolume).replaceAll(",", "") + ";");
                        outLine.append(volatilityFormat4.format(te.closeOpenPriceDiff)       + ";");
                        outLine.append(volatilityFormat4.format(te.highLowPriceDiff)         + ";");
                        outLine.append(volatilityFormat4.format(te.overnightPriceDiff)       + ";");
                        outLine.append(volatilityFormat4.format(te.adjCloseDiff)             + ";");
                        outLine.append(volatilityFormat4.format(te.volumeDiff)               + ";");
                        outLine.append(volatilityFormat4.format(te.dolVolumeDiff)            + ";");
                        eventWriter.newLine();
                        eventWriter.write(outLine.toString());
                    }
                }
                
                List<TradeEvent> seriesSublist = (List<TradeEvent>) truncateEventList(profileStartDateString, profileEndDateString, timeSeries);
                Profile profile = new Profile(seriesSublist);
                
                synchronized(profileWriter) {
                    StringBuilder writeThis = new StringBuilder(profile.symb + ";");
                    writeThis.append(profile.dateStart + ";");
                    writeThis.append(profile.dateEnd   + ";");
                    writeThis.append(priceFormat.format(profile.adjClose)             + ";");
                    writeThis.append(priceFormat.format(profile.adjCloseAvg)          + ";");
                    writeThis.append(priceFormat.format(profile.adjCloseMin)          + ";");
                    writeThis.append(priceFormat.format(profile.adjCloseMax)          + ";");
                    writeThis.append(priceFormat.format(profile.adjCloseMedian)       + ";");
                    writeThis.append(volatilityFormat4.format(profile.adjCloseStdDev) + ";");
                    writeThis.append(volumeFormat.format(profile.volumeAvg).replaceAll(",", "")    + ";");
                    writeThis.append(volumeFormat.format(profile.volumeMin).replaceAll(",", "")    + ";");
                    writeThis.append(volumeFormat.format(profile.volumeMax).replaceAll(",", "")    + ";");
                    writeThis.append(volumeFormat.format(profile.volumeMedian).replaceAll(",", "") + ";");
                    writeThis.append(volumeFormat.format(profile.volumeStdDev).replaceAll(",", "") + ";");
                    writeThis.append(volumeFormat.format(profile.dolVolumeAvg).replaceAll(",", "")    + ";");
                    writeThis.append(volumeFormat.format(profile.dolVolumeMin).replaceAll(",", "")    + ";");
                    writeThis.append(volumeFormat.format(profile.dolVolumeMax).replaceAll(",", "")    + ";");
                    writeThis.append(volumeFormat.format(profile.dolVolumeMedian).replaceAll(",", "") + ";");
                    writeThis.append(volumeFormat.format(profile.dolVolumeStdDev).replaceAll(",", "") + ";");
                    writeThis.append(volatilityFormat6.format(profile.adjCloseDiffAvg)          + ";");
                    writeThis.append(volatilityFormat6.format(profile.adjCloseDiffStdDev)       + ";");
                    writeThis.append(volatilityFormat6.format(profile.closeOpenPriceDiffAvg)    + ";");
                    writeThis.append(volatilityFormat6.format(profile.closeOpenPriceDiffStdDev) + ";");
                    writeThis.append(volatilityFormat6.format(profile.highLowPriceDiffAvg)      + ";");
                    writeThis.append(volatilityFormat6.format(profile.highLowPriceDiffStdDev)   + ";");
                    writeThis.append(volatilityFormat6.format(profile.overnightPriceDiffAvg)    + ";");
                    writeThis.append(volatilityFormat6.format(profile.overnightPriceDiffStdDev) + ";");
                    writeThis.append(volatilityFormat6.format(profile.volumeDiffAvg)            + ";");
                    writeThis.append(volatilityFormat6.format(profile.volumeDiffStdDev)         + ";");
                    writeThis.append(volatilityFormat6.format(profile.dolVolumeDiffAvg)         + ";");
                    writeThis.append(volatilityFormat6.format(profile.dolVolumeDiffStdDev)      + ";");
                    writeThis.append(productType + ";");
                    profileWriter.newLine();
                    profileWriter.write(writeThis.toString());
                }
            } catch(IOException e) {  }
            workCompleted++;
        }    
    }
    
    
    private List<TradeEvent> truncateEventList(String profileStartDate, String profileEndDate, List<TradeEvent> timeSeries) {
        List<TradeEvent> sublist = new ArrayList<TradeEvent>(timeSeries.size());
        try {
            SimpleDateFormat dateConverter2 = new SimpleDateFormat("yyyy-MM-dd");
            SimpleDateFormat dateConverter  = new SimpleDateFormat("MM-dd-yyyy");
            Date startDate = dateConverter.parse(profileStartDate);
            Date endDate   = dateConverter.parse(profileEndDate);
            int j = 0;
            
            for(int i = 0; i < timeSeries.size(); i++) {
                TradeEvent te = (TradeEvent) timeSeries.get(i);
                Date d = dateConverter2.parse(te.date);
                if((d.after(startDate) && d.before(endDate)) || d.equals(startDate) || d.equals(endDate)) {
                    sublist.add(j, te);
                    j++;
                }
            }
        } catch(ParseException e) { e.printStackTrace(); }
        return(sublist);
    }
 
    
    private URL createYahooURL(String symb) throws MalformedURLException {
        String url = ("http://ichart.finance.yahoo.com/table.csv?s=" + symb);
               url = (url + "&a=" + m1 + "&b=" + d1 + "&c=" + y1);
               url = (url + "&d=" + m2 + "&e=" + d2 + "&f=" + y2 + "&g=d");
        return(new URL(url));
    }
    
    
    public String toString() {
        NumberFormat f = NumberFormat.getInstance();
        f.setMaximumFractionDigits(2);
        f.setMinimumFractionDigits(2);
        
        double d = (double) ((((double) workCompleted) / ((double) initialAmountOfWork)) * 100);
        return("            worker #" + id + ":__number of downloads = " + workCompleted + ":__load balance = " + f.format(d) + "%");
    }
   
}